package com.atguigu.chapter07.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/15 11:01
 */
public class Flink01_State_Operator_List {
    public static void main(String[] args) {
        // 算子状态: 列表 联合列表 广播
        
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        // 默认情况, flink不会自动checkpoint. 需要开启
        env.enableCheckpointing(3000);
    
        env
            .socketTextStream("hadoop162", 9999)
            .flatMap(new MyFlatMapFunction())
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    static class MyFlatMapFunction implements FlatMapFunction<String, String>, CheckpointedFunction {
        ArrayList<String> words = new ArrayList<>();
        private ListState<String> listState;
    
        @Override
        public void flatMap(String line,
                            Collector<String> out) throws Exception {
            
            for (String word : line.split(" ")) {
                words.add(word);
                out.collect(words.toString());
                
            }
    
            if (line.contains("a")) {  // 为了演示状态的恢复
    
                throw new RuntimeException("手动抛异常");
            }
        
        }
    
        // 把状态进行快照(checkpoint), 其实就是把状态给保存下来(持久化)
        // 周期性的保存
        @Override
        public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
            System.out.println("MyFlatMapFunction.snapshotState");
            // 保存状态
            // 把ArrayList中的元素存储到列表状态中
            listState.update(words);
        }
    
        // 初始化 状态: 把持久化的状态,存入到内存状态中
        @Override
        public void initializeState(FunctionInitializationContext ctx) throws Exception {
            System.out.println("MyFlatMapFunction.initializeState");
            ListStateDescriptor<String> listStateDesc = new ListStateDescriptor<>("listSateDesc", String.class);
            listState = ctx.getOperatorStateStore().getListState(listStateDesc);
            // 把状态中的元素恢复到自创建的ArrayList
            for (String word : listState.get()) {
                System.out.println("恢复:" + word);
                words.add(word);
            }
        }
    }
}
